<?php


namespace App\Common\Kafka;


use RdKafka\Conf;
use RdKafka\Producer;

class KafkaProducer
{
    public static function produce(string $message = null)
    {
        $config = config('KafkaConfig.api');
        if (
            empty($config)
            || isset($config['bootstrap_servers']) == false
            || isset($config['topic_name']) == false
        ) {
            throw new \Exception('Kafka配置文件KafkaConfig.php错误');
        }
        $conf = new Conf();
        $conf->set('api.version.request', 'true');
        $rk = new Producer($conf);
        $rk->addBrokers(implode(',', $config['bootstrap_servers']));
        $topic = $rk->newTopic($config['topic_name']);
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
        $rk->flush(1000);
    }
}
